package org.apache.rocketmq.proxy.remoting;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.channel.Channel;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.future.FutureTaskExt;
import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
import org.apache.rocketmq.common.thread.ThreadPoolStatusMonitor;
import org.apache.rocketmq.common.utils.StartAndShutdown;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.remoting.activity.GetTopicRouteActivity;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.RemotingServer;
import org.apache.rocketmq.remoting.netty.NettyRemotingServer;
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.apache.rocketmq.remoting.netty.RequestTask;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;

import java.util.concurrent.*;

public class RemotingProtocolServer implements StartAndShutdown, RemotingProxyOutClient {
    private final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);

    //消息处理器，Netty服务器接收到的请求，到最后都会交给消息处理器处理
    protected final MessagingProcessor messagingProcessor;

    //客户端连接监听器，在第一、第二版本代码中，大家应该交过类似的监听器，也知道监听器是如何发挥作用的
    //但是在第三版本代码中，这个监听器主要负责监听生产者、消费者客户端和Proxy服务器的连接状态，也就是channel状态
    //但是我们现在根本没有引入生产者消费者，所以这里我就暂时把这个监听器设置为null，等后面引入了生产者消费者再给大家重构成源码的样子
    protected final ChannelEventListener clientHousekeepingService = null;

    //Netty服务器，这个类大家都已经见过了，在第一版本代码中，我就为大家引入了这个类
    protected final RemotingServer defaultRemotingServer;

    //根据主题获取路由信息的对象，这个对象是一个Netty的消息处理器，用来处理获取主题路由信息的请求
    protected final GetTopicRouteActivity getTopicRouteActivity;

    //getTopicRouteActivity这个消息处理器处理请求时用到的线程池
    protected final ThreadPoolExecutor topicRouteExecutor;

    //定时任务执行器，这个执行器会定期清理过期请求线程池中的过期请求
    protected final ScheduledExecutorService timerExecutor;


    public RemotingProtocolServer(MessagingProcessor messagingProcessor) {
        //得到消息处理器
        this.messagingProcessor = messagingProcessor;
        //创建根据主题获取路由信息的消息处理器
        this.getTopicRouteActivity = new GetTopicRouteActivity(messagingProcessor);
        //得到Proxy模块的配置信息对象
        ProxyConfig config = ConfigurationManager.getProxyConfig();
        //创建Netty服务器要使用的配置信息对象
        NettyServerConfig defaultServerConfig = new NettyServerConfig();
        //把Netty服务器要监听的本地端口号设置到配置信息对乡长
        defaultServerConfig.setListenPort(config.getRemotingListenPort());
        // 创建Netty服务器
        this.defaultRemotingServer = new NettyRemotingServer(defaultServerConfig, this.clientHousekeepingService);
        // 创建线程池
        this.topicRouteExecutor = ThreadPoolMonitor.createAndMonitor(
                config.getRemotingTopicRouteThreadPoolNums(),
                config.getRemotingTopicRouteThreadPoolNums(),
                1000 * 60, TimeUnit.SECONDS,
                "RemotingTopicRouteThread",
                config.getRemotingTopicRouteThreadPoolQueueCapacity(),
                new ThreadPoolHeadSlowTimeMillsMonitor(config.getRemotingWaitTimeMillsInTopicRouteQueue())
        );
        //注册 处理器
        registerRemotingServer(this.defaultRemotingServer);
        //创建单线程的定时任务执行器
        this.timerExecutor = ThreadUtils.newSingleThreadScheduledExecutor(
                new ThreadFactoryBuilder().setNameFormat("RemotingServerScheduler-%d").build()
        );
        this.timerExecutor.scheduleAtFixedRate(this::cleanExpireRequest,
                10,10,TimeUnit.SECONDS);


    }

    protected void cleanExpireRequest() {
        ProxyConfig config = ConfigurationManager.getProxyConfig();
        //从Proxy模块的配置信息对象中得到处理路由信息执行器的任务等待被执行的时间阈值
        cleanExpiredRequestInQueue(this.topicRouteExecutor, config.getRemotingWaitTimeMillsInTopicRouteQueue());
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/31
     * @方法描述：清除执行器中超时未被处理的任务的方法
     */
    protected void cleanExpiredRequestInQueue(ThreadPoolExecutor threadPoolExecutor, long maxWaitTimeMillsInQueue) {
        //在循环中不断判断是否有任务要被清除
        while (true) {
            try {
                //得到执行器的任务队列
                BlockingQueue<Runnable> blockingQueue = threadPoolExecutor.getQueue();
                if (!blockingQueue.isEmpty()) {
                    //的到任务队列中的第一个任务
                    final Runnable runnable = blockingQueue.peek();
                    if (null == runnable) {
                        break;
                    }
                    //如果任务不为空，则把任务转换为RequestTask类型的对象
                    final RequestTask rt = castRunnable(runnable);
                    if (rt == null || rt.isStopRun()) {
                        break;
                    }
                    //计算当前遍历到的任务的慢速时间，也就是等待被执行的时间
                    final long behind = System.currentTimeMillis() - rt.getCreateTimestamp();
                    //如果慢速时间超过了配置信息中的阈值，则把这个任务从队列中移除
                    if (behind >= maxWaitTimeMillsInQueue) {
                        //从任务队列中移除该任务，也就代表着Netty服务器不必处理这个请求了
                        if (blockingQueue.remove(runnable)) {
                            //设置任务停止标志
                            rt.setStopRun(true);
                            //给客户端回复系统繁忙的响应
                            rt.returnResponse(ResponseCode.SYSTEM_BUSY,
                                    String.format("[TIMEOUT_CLEAN_QUEUE]broker busy, start flow control for a while, period in queue: %sms, size of queue: %d", behind, blockingQueue.size()));
                        }
                    } else {
                        //队列中的第一个任务的慢速时间没有超时，那就直接退出循环，等待下一次定时任务执行的时候再检查
                        break;
                    }
                } else {
                    //如果任务队列为空，也直接退出循环
                    break;
                }
            } catch (Throwable ignored) {
                log.warn( "Unexpected exception occurred when waiting for completion {}",ignored);
            }
        }
    }

    protected class ThreadPoolHeadSlowTimeMillsMonitor implements ThreadPoolStatusMonitor {

        //慢速时间的阈值
        private final long maxWaitTimeMillsInQueue;

        //构造方法
        public ThreadPoolHeadSlowTimeMillsMonitor(long maxWaitTimeMillsInQueue) {
            this.maxWaitTimeMillsInQueue = maxWaitTimeMillsInQueue;
        }

        //返回当前监听器的描述信息
        @Override
        public String describe() {
            return "headSlow";
        }

        //返回当前线程池的头部任务的慢速时间
        @Override
        public double value(ThreadPoolExecutor executor) {
            return headSlowTimeMills(executor.getQueue());
        }

        //判断头部任务慢速时间是否超过了阈值，如果超过了阈值就需要记录栈信息了
        @Override
        public boolean needPrintJstack(ThreadPoolExecutor executor, double value) {
            return value > maxWaitTimeMillsInQueue;
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/31
     * @方法描述：计算执行器任务队列中第一个任务的慢速时间的方法
     */
    protected long headSlowTimeMills(BlockingQueue<Runnable> q) {
        try {
            long slowTimeMills = 0;
            //的到任务队列中的第一个任务
            final Runnable peek = q.peek();
            if (peek != null) {
                //把这个任务转换成RequestTask类型的对象，因为RequestTask对象中有该任务的创建时间
                RequestTask rt = castRunnable(peek);
                //计算该任务从创建到现在还未被执行等待的时间
                slowTimeMills = rt == null ? 0 : System.currentTimeMillis() - rt.getCreateTimestamp();
            }
            //为复数则置为0
            if (slowTimeMills < 0) {
                slowTimeMills = 0;
            }
            //返回任务队列第一个任务的慢速时间
            return slowTimeMills;
        } catch (Exception e) {
            log.error("error when headSlowTimeMills.", e);
        }
        return -1;
    }

    private RequestTask castRunnable(Runnable runnable) {
        try {
            if (runnable instanceof FutureTaskExt) {
                FutureTaskExt futureTaskExt = (FutureTaskExt) runnable;
                return (RequestTask) futureTaskExt.getRunnable();
            }
            return null;
        } catch (Throwable e) {
            log.error("castRunnable exception. class:{}", runnable.getClass().getName(), e);
        }

        return null;
    }

    protected void registerRemotingServer(RemotingServer remotingServer) {
        remotingServer.registerProcessor(RequestCode.GET_ROUTEINFO_BY_TOPIC, getTopicRouteActivity, this.topicRouteExecutor);
    }

    @Override
    public void shutdown() throws Exception {
        this.defaultRemotingServer.shutdown();
        this.topicRouteExecutor.shutdown();
    }

    @Override
    public void start() throws Exception {
        this.defaultRemotingServer.start();

    }

    @Override
    public CompletableFuture<RemotingCommand> invokeToClient(Channel channel, RemotingCommand request, long timeoutMillis) {
        return null;
    }
}
